package appTool

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"gitee.com/kinwyb/appTools/common"
	"gitee.com/kinwyb/appTools/log"
	clientv3 "go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc"
	"math/rand"
	"strings"
	"sync"
	"time"
)

// broadcast 用于广播消息内容，多个应用之间同步消息

// BroadcastMessage 广播消息结构体
type BroadcastMessage struct {
	ID          uint64 //消息唯一ID
	Sender      uint64
	MessageType uint8  //消息类型
	Message     []byte //消息内容
}

// Marshal 序列化消息
func (b BroadcastMessage) Marshal() []byte {
	id := make([]byte, 8)
	binary.BigEndian.PutUint64(id, b.ID)
	sender := make([]byte, 8)
	binary.BigEndian.PutUint64(sender, b.Sender)
	ret := append(id, sender...)
	ret = append(ret, b.MessageType)
	ret = append(ret, b.Message...)
	return ret
}

// UnMarshal 解析消息
func (b *BroadcastMessage) UnMarshal(data []byte) error {
	if len(data) < 17 {
		return errors.New("消息结构异常")
	}
	id := data[:8]
	sender := data[8:16]
	messageType := data[16:17]
	message := data[17:]
	b.ID = binary.BigEndian.Uint64(id)
	b.Sender = binary.BigEndian.Uint64(sender)
	b.MessageType = messageType[0]
	b.Message = message
	return nil
}

// BroadcastMessageCallback 消息广播回调结构,如果函数处理时间长会产生阻塞请注意处理
type BroadcastMessageCallback = func(msgs []BroadcastMessage)

// 基于etcd实现的消息广播
type etcdBroadcast struct {
	id                      uint64
	broadCastPath           string
	etcdClient              *clientv3.Client
	etcdHosts               []string
	mu                      sync.Mutex
	retriesAfterWatchFailed int // 失败后是否重试监听,-1表示一直重试直到监听成功, 0表示不重试.
	ctx                     context.Context
	cancelFunc              func()
	callback                []BroadcastMessageCallback
	msgIDGen                *common.EtcdIDGenerator
	timeout                 time.Duration
}

// newEtcdBroadcast returns a new newEtcdBroadcast.
func newEtcdBroadcast() *etcdBroadcast {
	d := &etcdBroadcast{retriesAfterWatchFailed: -1}
	d.msgIDGen = common.NewEtcdIDGenerator(uint16(rand.Uint32()>>16), time.Now())
	d.id = d.msgIDGen.Next()
	d.timeout = 10 * time.Second
	return d
}

func (d *etcdBroadcast) SetBasePath(basePath string) {
	if len(basePath) > 1 && strings.HasSuffix(basePath, "/") {
		basePath = basePath[:len(basePath)-1]
	}
	d.broadCastPath = basePath
}

func (d *etcdBroadcast) SetEtcdHosts(hosts []string) {
	d.etcdHosts = hosts
}

func (d *etcdBroadcast) Start(ctx context.Context) error {
	if len(d.etcdHosts) < 1 {
		return errors.New("etcd地址空")
	}
	if d.broadCastPath == "" {
		return errors.New("广播地址空")
	}
	d.ctx, d.cancelFunc = context.WithCancel(ctx)
	cfg := clientv3.Config{
		Endpoints:            d.etcdHosts,
		DialTimeout:          30 * time.Second,
		DialKeepAliveTimeout: time.Minute,
		DialKeepAliveTime:    15 * time.Second,
		DialOptions:          []grpc.DialOption{grpc.WithBlock()},
	}
	cli, err := clientv3.New(cfg)
	if err != nil {
		panic("etcd connect error:" + err.Error())
	}
	d.etcdClient = cli
	ctx, cancel := context.WithTimeout(context.Background(), d.timeout)
	resp, err := d.etcdClient.Get(ctx, d.broadCastPath, clientv3.WithPrefix())
	cancel()
	if err != nil {
		log.AppTool.WithField("info", d.broadCastPath).WithError(err).Error("无法获取到广播消息目录")
		return err
	}
	if len(resp.Kvs) > 0 {
		var waitNotifyMsg []BroadcastMessage
		var prefix string
		for _, v := range resp.Kvs {
			key := string(v.Key)
			if prefix == "" {
				if strings.HasPrefix(key, "/") {
					if strings.HasPrefix(d.broadCastPath, "/") {
						prefix = d.broadCastPath + "/"
					} else {
						prefix = "/" + d.broadCastPath + "/"
					}
				} else {
					if strings.HasPrefix(d.broadCastPath, "/") {
						prefix = d.broadCastPath[1:] + "/"
					} else {
						prefix = d.broadCastPath + "/"
					}
				}
			}
			if key == prefix[:len(prefix)-1] || !strings.HasPrefix(key, prefix) {
				continue
			}
			msg, e := d.filter(v.Value)
			if e != nil {
				continue
			}
			waitNotifyMsg = append(waitNotifyMsg, msg)
		}
		d.notifyWatcher(waitNotifyMsg)
	}
	go d.watch()
	return nil
}

func (d *etcdBroadcast) filter(value []byte) (BroadcastMessage, error) {
	msg := BroadcastMessage{}
	if len(value) < 1 {
		return msg, errors.New("无效消息")
	}
	err := (&msg).UnMarshal(value)
	if err != nil {
		return msg, err
	}
	return msg, nil
}

// 通知消息监听者
func (d *etcdBroadcast) notifyWatcher(msgs []BroadcastMessage) {
	if len(d.callback) < 1 || len(msgs) < 1 {
		return
	}
	var notifyMsg = make([]BroadcastMessage, len(msgs))
	var i = 0
	for _, v := range msgs {
		if v.ID == 0 || v.Sender == d.id {
			continue
		}
		notifyMsg[i] = v
		i++
	}
	if i > 0 {
		notifyMsg = notifyMsg[:i]
		d.mu.Lock()
		for _, v := range d.callback {
			v(notifyMsg)
		}
		d.mu.Unlock()
	}
}

// SendMessage 广播一条消息
func (d *etcdBroadcast) SendMessage(messageType uint8, message []byte) error {
	msg := BroadcastMessage{
		ID:          d.msgIDGen.Next(),
		Sender:      d.id,
		MessageType: messageType,
		Message:     message,
	}
	key := fmt.Sprintf("%s/%d", d.broadCastPath, msg.ID)
	leaseResp, err := d.etcdClient.Grant(context.Background(), 5)
	if err != nil {
		log.AppTool.WithError(err).Error("广播消息租约申请失败")
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), d.timeout)
	_, err = d.etcdClient.Put(ctx, key, string(msg.Marshal()), clientv3.WithLease(leaseResp.ID))
	cancel()
	return err
}

// SetRetriesAfterWatchFailed 添加系统消息回调用
func (d *etcdBroadcast) SetRetriesAfterWatchFailed(retryNum int) {
	d.retriesAfterWatchFailed = retryNum
}

// AddCallback 添加系统消息回调用
func (d *etcdBroadcast) AddCallback(callback BroadcastMessageCallback) {
	d.mu.Lock()
	defer d.mu.Unlock()
	callbackPointer := fmt.Sprintf("%p", callback)
	for _, v := range d.callback {
		if fmt.Sprintf("%p", v) == callbackPointer {
			return
		}
	}
	d.callback = append(d.callback, callback)
}

func (d *etcdBroadcast) RemoveCallback(callback BroadcastMessageCallback) {
	d.mu.Lock()
	defer d.mu.Unlock()
	var callbacks []BroadcastMessageCallback
	callbackPointer := fmt.Sprintf("%p", callback)
	for _, v := range d.callback {
		if fmt.Sprintf("%p", v) == callbackPointer {
			continue
		}
		callbacks = append(callbacks, v)
	}
	d.callback = callbacks
}

func (d *etcdBroadcast) watch() {
	var retryNum = d.retriesAfterWatchFailed
	for {
		var tempDelay time.Duration
		var watchChan clientv3.WatchChan
		if d.retriesAfterWatchFailed < 0 || retryNum >= 0 {
			ctx, cancel := context.WithCancel(d.ctx)
			watchChan = d.etcdClient.Watch(ctx, d.broadCastPath, clientv3.WithPrefix())
			endLoop := false
		retryWatch:
			for !endLoop {
				select {
				case <-d.ctx.Done():
					cancel() //关闭watch
					log.AppTool.Debug("广播消息监听结束")
					return
				case c := <-watchChan:
					if c.Err() != nil {
						log.AppTool.WithError(c.Err()).Debug("广播监听失败")
						retryNum--
						endLoop = true
						break retryWatch
					}
					if len(c.Events) == 0 {
						break
					}
					retryNum = d.retriesAfterWatchFailed //成功后重新计数
					var pairs []BroadcastMessage         // new messages
					for _, event := range c.Events {
						if event.IsModify() || event.IsCreate() {
							key := string(event.Kv.Key)
							var prefix string
							if strings.HasPrefix(key, "/") {
								if strings.HasPrefix(d.broadCastPath, "/") {
									prefix = d.broadCastPath + "/"
								} else {
									prefix = "/" + d.broadCastPath + "/"
								}
							} else {
								if strings.HasPrefix(d.broadCastPath, "/") {
									prefix = d.broadCastPath[1:] + "/"
								} else {
									prefix = d.broadCastPath + "/"
								}
							}
							if key == prefix[:len(prefix)-1] || !strings.HasPrefix(key, prefix) {
								continue
							}
							msg, err := d.filter(event.Kv.Value)
							if err != nil {
								continue
							}
							pairs = append(pairs, msg)
						}
					}
					d.notifyWatcher(pairs)
				}
			}
			cancel() //关闭watch
			if tempDelay == 0 {
				tempDelay = 1 * time.Second
			} else {
				tempDelay *= 2
			}
			if max := 30 * time.Second; tempDelay > max {
				tempDelay = max
			}
			time.Sleep(tempDelay)
		} else {
			log.AppTool.Error("etcdBroadcast watch fail")
		}
		log.AppTool.Warn("etcdBroadcast chan is closed and will rewatch")
	}
}

func (d *etcdBroadcast) Close() {
	if d.cancelFunc != nil {
		d.cancelFunc()
	}
}
